package com.atguigu.chapter04;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.stream.Stream;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/2 10:01
 */
public class Flink01_Parallelism {
    public static void main(String[] args) throws Exception {
        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(3);

        // 2.读取数据
//        DataStreamSource<String> socketDS = env.socketTextStream("localhost", 9999);
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop1", 9999);

        // 3.处理数据
        // 3.1 压平：切分
        SingleOutputStreamOperator<String> wordDS = socketDS
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        String[] words = value.split(" ");
                        for (String word : words) {
                            out.collect(word);
                        }
                    }
                })
                .setParallelism(4);
        // 3.2 转换成二元组（word，1）
        SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOneDS = wordDS
                .map(value -> Tuple2.of(value, 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG));

        // 3.3 按照word分组
        KeyedStream<Tuple2<String, Long>, Tuple> wordAndOneKS = wordAndOneDS.keyBy(0);

        // 3.4 组内求和
        SingleOutputStreamOperator<Tuple2<String, Long>> resultDS = wordAndOneKS.sum(1);

        // 4. 输出
        resultDS.print();

        // 5. 启动
        env.execute();

    }


}

/*
    并行度优先级：  代码的算子指定 > 代码的env（全局指定） > 提交参数 > 配置文件默认

    并行度与slot的关系：
        1）同一个算子的子任务，不能在同一个slot里
        2）不同算子的子任务，可以共享一个slot
        3）如果 slot数量 < 算子最大并行度，
            如果是 Standalone模式，那么会卡住，等待资源足够
            如果是 Yarn模式，会动态申请更多资源

    整个流式程序的并行度  =  算子的最大并行度
 */


/*

    SubTask： 一个算子的 一个 并行实例

    Task: 一个算子的一个子任务， 或 多个算子的子任务 满足某种不可描述的关系（one-to-one，并行度相同，经过 Operator Chain优化），串在一起之后的 子任务

    举例：sum算子 跟 print算子 并行度都是3


    sum算子   print算子
    ⭕           ⭕
    ⭕           ⭕
    ⭕           ⭕

    sum和print满足某种不可描述的关系，他们之间的子任务可以 串在一起

    sum算子   print算子
    （⭕           ⭕） => 新的 subtask  => 执行时的 一个 Task
    （⭕           ⭕） => 新的 subtask  => 执行时的 一个 Task
    （⭕           ⭕） => 新的 subtask  => 执行时的 一个 Task

 */